package com.shujia.flink.core

import org.apache.flink.streaming.api.scala._

object Demo2RpcSubmit {
  def main(args: Array[String]): Unit = {

    /**
      * 创建远程环境，远程提交flink任务
      *
      */
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createRemoteEnvironment(
      "node1",
      38850,
      "C:\\Users\\qx\\IdeaProjects\\bigdata14\\flink\\target\\flink-1.0.jar"
    )


    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)
    //1、将数据展开
    val wordsDS: DataStream[String] = linesDS.flatMap(line => line.split(","))
    //2、转换成kv格式
    val kvDS: DataStream[(String, Int)] = wordsDS.map((_, 1))
    //3、按照单词进行分组
    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(kv => kv._1)
    //4、统计数量,对value进行求和, 指定下标进行聚合
    val countDS: DataStream[(String, Int)] = keyByDS.sum(1)
    //打印结果
    countDS.print()

    env.execute("rpc")


  }

}
